package com.lagou.mall.esflow.server.handler;

import com.lagou.mall.esflow.client.AckEnum;
import com.lagou.mall.esflow.client.Binlog;
import com.lagou.mall.esflow.server.config.EnvConfig;
import com.lagou.mall.esflow.server.util.KafkaMsgUtil;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.KafkaConsumer;

public abstract class AbstractTopicConsumerHandler {


    public static final String BIN_LOG_ID = "id";
    public static final String UNDER_SCORE = "_";

    public abstract AckEnum start(KafkaConsumer consumer, Binlog binlog);


    public String getId(Binlog binlog) {
        return String.valueOf(KafkaMsgUtil.getKey(BIN_LOG_ID, binlog));
    }


    public String getTable(Binlog binlog) {
        return binlog.getTable();
    }


    public String getValueByField(String field, Binlog binlog) {
        return String.valueOf(KafkaMsgUtil.getKey(field, binlog));

    }

    public String assembleIndex(EnvConfig envConfig, String originIndex) {
        if (envConfig.isSit() || envConfig.isTest()) {
            return envConfig.getEnv() + UNDER_SCORE + originIndex;
        }
        return originIndex;
    }

}
